package com.shujia.flink.sql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo11EventTime {
  def main(args: Array[String]): Unit = {
    //创建flink的环境
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 修改事件模式为事件时间
      *
      */

    bsEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //流模式
      .build()

    /**
      * 创建flink sql 环境
      *
      */

    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)


    /*


java,"2022-01-24 14:37:10"
java,"2022-01-24 14:37:11"
java,"2022-01-24 14:37:12"
java,"2022-01-24 14:37:13"
java,"2022-01-24 14:37:14"
java,"2022-01-24 14:37:15"
java,"2022-01-24 14:37:16"
java,"2022-01-24 14:37:17"
java,"2022-01-24 14:37:18"
java,"2022-01-24 14:37:20"


     */
    bsTableEnv.executeSql(
      """
        |
        |CREATE TABLE words_event_time (
        |  word STRING,
        |  user_action_time TIMESTAMP(3),
        |  -- 声明 user_action_time 是事件时间属性，并且用 延迟 5 秒的策略来生成 watermark
        |  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
        |) WITH (
        |'connector' = 'kafka',
        | 'topic' = 'words_event_time',
        | 'properties.bootstrap.servers' = 'master:9092',
        | 'properties.group.id' = 'asdasd',
        | 'format' = 'csv',
        | 'csv.field-delimiter' = ',',
        | 'csv.ignore-parse-errors' = 'true'
        |)
        |
        |
      """.stripMargin)



    /**
      * 创建sink表保存数据
      */


    bsTableEnv.executeSql(
      """
        |CREATE TABLE mysql_sink (
        |  word STRING,
        |  window_end TIMESTAMP(3),
        |  c BIGINT,
        |  PRIMARY KEY (word,window_end) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/bigdata',
        |   'table-name' = 'word_window_count',
        |   'username'='root',
        |   'password'='123456'
        |)
        |
        |
      """.stripMargin)

    /**
      * 每隔5秒统计单词的数量
      *
      */

    bsTableEnv.executeSql(
      """
        |
        |insert into mysql_sink
        |select
        | word,
        | TUMBLE_END(user_action_time, INTERVAL '5' SECOND) as window_end,
        | count(1) as c
        |from
        | words_event_time
        |group by
        | word,
        | TUMBLE(user_action_time, INTERVAL '5' SECOND)
        |
      """.stripMargin)



  }

}
